package cn.chengpengper.im.sdk.interfaces.sender.impl;

import cn.chengpengper.im.common.cache.distribute.DistributedCacheService;
import cn.chengpengper.im.common.domain.constant.ImConstants;
import cn.chengpengper.im.common.domain.enums.IMTerminalType;
import cn.chengpengper.im.common.domain.enums.ImCmdType;
import cn.chengpengper.im.common.domain.enums.ImListenerType;
import cn.chengpengper.im.common.domain.enums.ImSendCode;
import cn.chengpengper.im.common.domain.model.*;
import cn.chengpengper.im.common.mq.MessageSenderService;
import cn.chengpengper.im.sdk.infrastructure.multicaster.MessageListenerMulticaster;
import cn.chengpengper.im.sdk.interfaces.sender.ImSender;
import cn.hutool.core.util.BooleanUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.collections.MapUtils;
import org.apache.commons.lang3.StringUtils;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.*;

@Slf4j
@Component
public class DefaultImSender implements ImSender {

    @Resource
    private DistributedCacheService distributedCacheService;

    @Resource
    private MessageSenderService messageSenderService;

    @Resource
    private MessageListenerMulticaster messageListenerMulticaster;

    @Override
    public <T> void sendPrivateMessage(ImPrivateMessage<T> message) {
        if (Objects.isNull(message)) {
            return;
        }

        List<Integer> receiveTerminals = message.getReceiveTerminals();
        if (CollectionUtils.isNotEmpty(receiveTerminals)) {
            return;
        }
        // 向目标用户发送私聊消息
        sendPrivateMessageToTargetUser(message, receiveTerminals);
        // 向自己的其他终端发送私聊消息
        sendPrivateMessageToSelf(message, receiveTerminals);
    }

    @Override
    public <T> void sendGroupMessage(ImGroupMessage<T> message) {
        Map<String, ImUserInfo> userTerminalGroup = getUserTerminalGroup(message);
        if (MapUtils.isEmpty(userTerminalGroup)) {
            return;
        }

        List<String> serverIdList = distributedCacheService.multiGet(userTerminalGroup.keySet());
        if (CollectionUtils.isEmpty(serverIdList)) {
            return;
        }
        Map<Integer, List<ImUserInfo>> serverMap = new HashMap<>();
        List<ImUserInfo> offlineUserList = new LinkedList<>();
        int idx = 0;
        for (Map.Entry<String, ImUserInfo> entry : userTerminalGroup.entrySet()) {
            String serverId = serverIdList.get(idx++);
            if (StringUtils.isNotBlank(serverId)) {
                List<ImUserInfo> list = serverMap.computeIfAbsent(Integer.parseInt(serverId), o -> new LinkedList<>());
                list.add(entry.getValue());
            } else {
                offlineUserList.add(entry.getValue());
            }
        }
        sendGroupMessageToOtherUsers(serverMap, offlineUserList, message);
        sendGroupMessageToSelf(message);
    }

    @Override
    public Map<Long, List<IMTerminalType>> getOnlineTerminal(List<Long> userIds) {
        if (CollectionUtils.isEmpty(userIds)) {
            return Collections.emptyMap();
        }

        Map<String, ImUserInfo> userMap = new HashMap<>();
        for (Long userId : userIds) {
            for (Integer terminal : IMTerminalType.codes()) {
                // 在用户登录的时候登录处理器（LoginProcessor）会记录到缓存中，KEY:用户id+终端类型 Value:serverId
                String key = String.join(ImConstants.REDIS_KEY_SPLIT, ImConstants.IM_USER_SERVER_ID, userId.toString(), terminal.toString());
                userMap.put(key, new ImUserInfo(userId, terminal));
            }
        }
        // 从Redis批量获取数据——服务ID
        List<String> serverIdList = distributedCacheService.multiGet(userMap.keySet());
        int idx = 0;
        Map<Long, List<IMTerminalType>> onlineMap = new HashMap<>();
        for (Map.Entry<String, ImUserInfo> entry : userMap.entrySet()) {
            if (StringUtils.isNotBlank(serverIdList.get(idx++))) {
                ImUserInfo imUserInfo = entry.getValue();
                List<IMTerminalType> imTerminalTypes = onlineMap.computeIfAbsent(imUserInfo.getUserId(), o -> new ArrayList<>());
                imTerminalTypes.add(IMTerminalType.fromCode(imUserInfo.getTerminal()));
            }
        }

        return onlineMap;
    }

    @Override
    public Boolean isOnline(Long userId) {
        String redisKey = String.join(ImConstants.REDIS_KEY_SPLIT, ImConstants.IM_USER_SERVER_ID, userId.toString(), "*");
        Set<String> keys = distributedCacheService.keys(redisKey);
        return CollectionUtils.isNotEmpty(keys);
    }

    @Override
    public List<Long> getOnlineUser(List<Long> userIds) {
        return new ArrayList<>(getOnlineTerminal(userIds).keySet());
    }

    private <T> void sendGroupMessageToSelf(ImGroupMessage<T> message) {
        for (Integer terminal : IMTerminalType.codes()) {
            if (terminal.equals(message.getSender().getTerminal())) {
                continue;
            }
            String redisKey = String.join(ImConstants.REDIS_KEY_SPLIT, ImConstants.IM_USER_SERVER_ID, message.getSender().getUserId().toString(), terminal.toString());
            String serverId = distributedCacheService.get(redisKey);
            if (StringUtils.isNotBlank(serverId)) {
                ImReceiveInfo imReceiveInfo = new ImReceiveInfo(
                        ImCmdType.GROUP_MESSAGE.getCode(),
                        message.getSender(),
                        Collections.singletonList(new ImUserInfo(message.getSender().getUserId(), terminal)),
                        false,
                        message.getData()
                );
                String sendKey = String.join(ImConstants.MESSAGE_KEY_SPLIT, ImConstants.IM_MESSAGE_GROUP_QUEUE, serverId);
                imReceiveInfo.setDestination(sendKey);
                messageSenderService.send(imReceiveInfo);
            }
        }
    }

    /**
     * 向群组其他成员发送消息
     *
     * @param serverMap
     * @param offlineUserList
     * @param message
     * @param <T>
     */
    private <T> void sendGroupMessageToOtherUsers(Map<Integer, List<ImUserInfo>> serverMap, List<ImUserInfo> offlineUserList, ImGroupMessage<T> message) {
        for (Map.Entry<Integer, List<ImUserInfo>> entry : serverMap.entrySet()) {
            ImReceiveInfo imReceiveInfo = new ImReceiveInfo(
                    ImCmdType.GROUP_MESSAGE.getCode(),
                    message.getSender(),
                    new LinkedList<>(entry.getValue()),
                    message.getSendResult(),
                    message.getData()
            );
            String sendKey = String.join(ImConstants.MESSAGE_KEY_SPLIT, ImConstants.IM_MESSAGE_GROUP_QUEUE, entry.getKey().toString());
            imReceiveInfo.setDestination(sendKey);
            messageSenderService.send(imReceiveInfo);
        }
        if (message.getSendResult()) {
            offlineUserList.forEach((offlineUser) -> {
                ImSendResult<T> result = new ImSendResult<>(
                        message.getSender(),
                        offlineUser,
                        ImSendCode.NOT_ONLINE.getCode(),
                        message.getData()
                );
                messageListenerMulticaster.multicast(ImListenerType.GROUP_MESSAGE, result);
            });
        }
    }

    private <T> Map<String, ImUserInfo> getUserTerminalGroup(ImGroupMessage<T> message) {
        Map<String, ImUserInfo> map = new HashMap<>(16);
        if (Objects.isNull(message)) {
            return map;
        }

        for (Integer terminal : message.getReceiveTerminals()) {
            message.getReceiveIds().forEach((receiveId) -> {
                String key = String.join(ImConstants.REDIS_KEY_SPLIT, ImConstants.IM_USER_SERVER_ID, receiveId.toString(), terminal.toString());
                map.put(key, new ImUserInfo(receiveId, terminal));
            });
        }
        return map;
    }

    private <T> void sendPrivateMessageToSelf(ImPrivateMessage<T> message, List<Integer> receiveTerminals) {
        if (BooleanUtil.isTrue(message.getSendSelf())) {
            receiveTerminals.forEach((receiveTerminal) -> {
                if (!receiveTerminal.equals(message.getSender().getTerminal())) {
                    String redisKey = String.join(ImConstants.REDIS_KEY_SPLIT, ImConstants.IM_USER_SERVER_ID, message.getSender().getUserId().toString(), receiveTerminal.toString());
                    String serverId = distributedCacheService.get(redisKey);
                    if (StringUtils.isNotBlank(serverId)) {
                        String sendKey = String.join(ImConstants.MESSAGE_KEY_SPLIT, ImConstants.IM_MESSAGE_PRIVATE_QUEUE, serverId);
                        ImReceiveInfo imReceiveInfo = new ImReceiveInfo(
                                ImCmdType.PRIVATE_MESSAGE.getCode(),
                                message.getSender(),
                                Collections.singletonList(new ImUserInfo(message.getReceiveId(), receiveTerminal)),
                                message.getSendResult(),
                                message.getData());

                        imReceiveInfo.setDestination(sendKey);
                        messageSenderService.send(imReceiveInfo);
                    }
                }
            });
        }
    }


    private <T> void sendPrivateMessageToTargetUser(ImPrivateMessage<T> message, List<Integer> receiveTerminals) {
        receiveTerminals.forEach((receiveTerminal) -> {
            String redisKey = String.join(ImConstants.REDIS_KEY_SPLIT, ImConstants.IM_USER_SERVER_ID, message.getReceiveId().toString(), receiveTerminal.toString());
            String serverId = distributedCacheService.get(redisKey);
            if (StringUtils.isNotBlank(serverId)) {
                // 用户在线，直接推送到MQ
                String sendKey = String.join(ImConstants.MESSAGE_KEY_SPLIT, ImConstants.IM_MESSAGE_PRIVATE_QUEUE, serverId);
                ImReceiveInfo imReceiveInfo = new ImReceiveInfo(
                        ImCmdType.PRIVATE_MESSAGE.getCode(),
                        message.getSender(),
                        Collections.singletonList(new ImUserInfo(message.getReceiveId(), receiveTerminal)),
                        message.getSendResult(),
                        message.getData());

                imReceiveInfo.setDestination(sendKey);
                messageSenderService.send(imReceiveInfo);
            } else if (BooleanUtil.isTrue(message.getSendResult())) {
                //回复消息的状态
                ImSendResult<T> result = new ImSendResult<>(message.getSender(), new ImUserInfo(message.getReceiveId(), receiveTerminal), ImSendCode.NOT_ONLINE.getCode(), message.getData());
                messageListenerMulticaster.multicast(ImListenerType.PRIVATE_MESSAGE, result);
            }
        });
    }
}
